/**
 * $Revision: $
 * $Date: $
 *
 * Copyright (C) 2005-2008 Jive Software. All rights reserved.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package net.lastcoder.ifserver.nio;

import net.lastcoder.ifserver.Server;
import net.lastcoder.ifserver.packet.Packet;
import net.lastcoder.ifserver.session.Session;

import org.apache.mina.core.buffer.IoBuffer;
import org.apache.mina.core.session.IoSession;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.net.SocketAddress;
import java.net.UnknownHostException;
import java.nio.charset.CharsetEncoder;
import java.nio.charset.CodingErrorAction;
import java.nio.charset.StandardCharsets;
import java.util.Arrays;
import java.util.concurrent.atomic.AtomicReference;
import java.util.concurrent.locks.ReentrantLock;

public class NIOConnection implements Connection {
	private static final Logger Log = LoggerFactory.getLogger(NIOConnection.class);

    private Session session;
    private IoSession ioSession;

    private ConnectionCloseListener closeListener;

    private AtomicReference<State> state = new AtomicReference<>(State.OPEN);
    private final ReentrantLock ioSessionLock = new ReentrantLock(true);

    public NIOConnection(IoSession session) {
        this.ioSession = session;
    }

    @Override
    public Session getOwner() {
        return session;
    }

    @Override
    public boolean validate() {
        if (isClosed()) {
            return false;
        }
        deliverRawData(" ".getBytes());
        return !isClosed();
    }

    @Override
    public void registerCloseListener(ConnectionCloseListener listener, Object ignore) {
        if (closeListener != null) {
            throw new IllegalStateException("Close listener already configured");
        }
        if (isClosed()) {
            listener.onConnectionClose(session);
        }
        else {
            closeListener = listener;
        }
    }

    @Override
    public void removeCloseListener(ConnectionCloseListener listener) {
        if (closeListener == listener) {
            closeListener = null;
        }
    }

    @Override
    public byte[] getAddress() throws UnknownHostException {
        final SocketAddress remoteAddress = ioSession.getRemoteAddress();
        if (remoteAddress == null) throw new UnknownHostException();
        final InetSocketAddress socketAddress = (InetSocketAddress) remoteAddress;
        final InetAddress address = socketAddress.getAddress();
        return address.getAddress();
    }

    @Override
    public String getHostAddress() throws UnknownHostException {
        final SocketAddress remoteAddress = ioSession.getRemoteAddress();
        if (remoteAddress == null) throw new UnknownHostException();
        final InetSocketAddress socketAddress = (InetSocketAddress) remoteAddress;
        final InetAddress inetAddress = socketAddress.getAddress();
        return inetAddress.getHostAddress();
    }

    @Override
    public String getHostName() throws UnknownHostException {
        final SocketAddress remoteAddress = ioSession.getRemoteAddress();
        if (remoteAddress == null) throw new UnknownHostException();
        final InetSocketAddress socketAddress = (InetSocketAddress) remoteAddress;
        final InetAddress inetAddress = socketAddress.getAddress();
        return inetAddress.getHostName();
    }

    @Override
    public void close() {
    	if (state.compareAndSet(State.OPEN, State.CLOSED)) {

    		if (session != null) {
                session.setStatus( Session.STATUS_CLOSED );
            }

            try {
                // TODO 关闭连接消息
                IoBuffer buffer = IoBuffer.allocate(1);
                buffer.setAutoExpand(false);
                byte b = '\0';
                buffer.put(b);

                buffer.flip();
                ioSessionLock.lock();
                try {
                    ioSession.write(buffer);
                }
                finally {
                    ioSessionLock.unlock();
                }
            } catch (Exception e) {
                Log.error("Failed to deliver stream close tag: " + e.getMessage(), e);
            }

            try {
                ioSession.close(true);
            } catch (Exception e) {
                Log.error("Exception while closing MINA session", e);
            }

            notifyCloseListeners(); // clean up session, etc.
        }
    }

    @Override
    public void systemShutdown() {
        // 关闭连接消息
        close();
    }

    private void notifyCloseListeners() {
        if (closeListener != null) {
            try {
                closeListener.onConnectionClose(session);
            } catch (Exception e) {
                Log.error("Error notifying listener: " + closeListener, e);
            }
        }
    }

    @Override
    public void init(Session owner) {
        session = owner;
    }

    @Override
    public boolean isClosed() {
    	return state.get() == State.CLOSED;
    }


    @Override
    public void deliver(Packet packet) {
        if (packet == null) {
            return;
        }
        if (!isClosed()) {
            boolean errorDelivering = false;
            try {
                ioSessionLock.lock();
                try {
                    IoBuffer buffer = packet.getIoBuffer();
                    
                    int len = buffer.capacity();
                    Server.outConnCount += len;
                    Server.outCount += len;
                    
                    ioSession.write(packet.getIoBuffer());
                } finally {
                    ioSessionLock.unlock();
                }
            }
            catch (Exception e) {
                Log.error("Error delivering Packet:\n" + packet, e);
                errorDelivering = true;
            }
            if (errorDelivering) {
                close();
            }
        }
    }

    @Override
    public void deliverRawData(byte[] data) {
        if (!isClosed()) {
            boolean errorDelivering = false;
            IoBuffer buffer = IoBuffer.allocate(data.length);
            try {
                buffer.put(data);
                buffer.flip();
                
                int len = data.length;
                Server.outConnCount += len;
                Server.outCount += len;
                
                ioSessionLock.lock();
                try {
                    ioSession.write(buffer);
                }
                finally {
                    ioSessionLock.unlock();
                }
            }
            catch (Exception e) {
                Log.error("Error delivering raw data:\n" + Arrays.toString(data), e);
                errorDelivering = true;
            }

            // Attempt to close the connection if delivering text fails.
            if (errorDelivering) {
                close();
            }
        }
    }

    @Override
    public void deliverBuffer(IoBuffer buffer) {
        if (!isClosed()) {
            boolean errorDelivering = false;
            try {
                
                int len = buffer.capacity();
                Server.outConnCount += len;
                Server.outCount += len;
                
                ioSessionLock.lock();
                try {
                    ioSession.write(buffer);
                }
                finally {
                    ioSessionLock.unlock();
                }
            }
            catch (Exception e) {
                Log.error("Error delivering raw data:\n" + Arrays.toString(buffer.array()), e);
                errorDelivering = true;
            }

            // Attempt to close the connection if delivering text fails.
            if (errorDelivering) {
                close();
            }
        }
    }

    @Override
    public void deliverRawData(byte[] data, int off, int len) {
        if (!isClosed()) {
            boolean errorDelivering = false;
            IoBuffer buffer = IoBuffer.allocate(data.length);
            buffer.setAutoExpand(true);
            try {
                buffer.put(data, off, len);
                
                Server.outConnCount += len;
                Server.outCount += len;
                
                buffer.flip();
                ioSessionLock.lock();
                try {
                    ioSession.write(buffer);
                }
                finally {
                    ioSessionLock.unlock();
                }
            }
            catch (Exception e) {
                Log.error("Error delivering raw data:\n" + Arrays.toString(data), e);
                errorDelivering = true;
            }

            // Attempt to close the connection if delivering text fails.
            if (errorDelivering) {
                close();
            }
        }
    }

    private static class ThreadLocalEncoder extends ThreadLocal<CharsetEncoder> {
        @Override
		protected CharsetEncoder initialValue() {
            return StandardCharsets.UTF_8.newEncoder()
				.onMalformedInput(CodingErrorAction.REPORT)
				.onUnmappableCharacter(CodingErrorAction.REPORT);
        }
    }

    @Override
    public String toString() {
        return "NIOConnection{" +
                "ioSession=" + ioSession +
                ", state=" + state +
                '}';
    }
}
